//! WebSocket客户端
//! 
//! 提供统一的WebSocket连接和消息处理

use crate::config::BinanceConfig;
use crate::errors::{BinanceError, Result};
use crate::websocket::models::*;
use futures_util::{SinkExt, StreamExt};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio_native_tls::native_tls;
use tokio_native_tls::TlsConnector;
use tokio_tungstenite::{client_async, connect_async, WebSocketStream, tungstenite::Message};
use tungstenite::handshake::client::Response;
use url::Url;

/// WebSocket客户端
pub struct WebSocketClient {
    config: Arc<BinanceConfig>,
    sender: Option<mpsc::UnboundedSender<WebSocketMessage>>,
    receiver: Option<mpsc::UnboundedReceiver<WebSocketMessage>>,
    streams: Vec<String>,
}

/// WebSocket消息类型
#[derive(Debug, Clone)]
pub enum WebSocketMessage {
    /// 订阅流
    Subscribe(String),
    /// 取消订阅
    Unsubscribe(String),
    /// 关闭连接
    Close,
}

impl WebSocketClient {
    /// 创建新的WebSocket客户端
    pub fn new(config: Arc<BinanceConfig>) -> Result<Self> {
        let (sender, receiver) = mpsc::unbounded_channel();
        
        Ok(Self {
            config,
            sender: Some(sender),
            receiver: Some(receiver),
            streams: Vec::new(),
        })
    }

    /// 连接到WebSocket
    pub async fn connect(&mut self) -> Result<()> {
        let base_url = self.config.ws_base_url();
        let url = base_url.to_string(); // 直接使用base_url，不添加/ws后缀
        
        tracing::info!("[WEBSOCKET] 连接URL: {}", url);
        tracing::info!("[WEBSOCKET] 代理配置: {:?}", self.config.ws_proxy);
        tracing::info!("[WEBSOCKET] HTTP代理: {:?}", self.config.http_proxy);
        
        // 对于Binance WebSocket API，需要构建正确的流URL
        // 根据Binance官方文档，WebSocket URL应该是这样的格式：
        // wss://stream.binance.com:9443/ws/btcusdt@ticker
        // 或者 wss://stream.binance.com:9443/stream?streams=btcusdt@ticker
        
        let ws_url = if let Some(proxy) = &self.config.ws_proxy {
            // 如果有WebSocket代理，需要特殊处理
            tracing::info!("[WEBSOCKET] 使用WebSocket代理: {}", proxy);
            tracing::info!("[WEBSOCKET] 目标URL: {}", url);
            url // 代理信息不应该添加到URL中
        } else {
            tracing::info!("[WEBSOCKET] 无代理连接");
            url
        };

        let url = Url::parse(&ws_url)
            .map_err(|e| BinanceError::ConfigError(format!("Invalid WebSocket URL: {}", e)))?;

        tracing::info!("[WEBSOCKET] 解析后的URL: {}", url);
        tracing::info!("[WEBSOCKET] 开始连接...");
        
        // 根据是否有代理选择不同的连接方式
        let ws_stream = if let Some(proxy) = &self.config.ws_proxy {
            tracing::info!("[WEBSOCKET] 通过代理连接: {}", proxy);
            connect_through_proxy(&url, proxy).await?
        } else {
            tracing::info!("[WEBSOCKET] 直接连接");
            connect_direct(&url).await?
        };
        
        tracing::info!("[WEBSOCKET] 连接成功！");

        let (mut write, mut read) = ws_stream.split();
        
        // 等待连接确认
        tracing::info!("WebSocket连接已建立，等待连接确认...");
        sleep(Duration::from_millis(100)).await;

        // 启动消息处理循环
        let mut receiver = self.receiver.take().unwrap();
        let _streams = Arc::new(tokio::sync::Mutex::new(self.streams.clone()));

        tokio::spawn(async move {
            while let Some(message) = receiver.recv().await {
                match message {
                    WebSocketMessage::Subscribe(stream) => {
                        let subscribe_msg = serde_json::json!({
                            "method": "SUBSCRIBE",
                            "params": [stream],
                            "id": 1
                        });
                        
                        if let Err(e) = write.send(Message::Text(subscribe_msg.to_string())).await {
                            eprintln!("Failed to send subscribe message: {}", e);
                            break;
                        }
                    }
                    WebSocketMessage::Unsubscribe(stream) => {
                        let unsubscribe_msg = serde_json::json!({
                            "method": "UNSUBSCRIBE",
                            "params": [stream],
                            "id": 1
                        });
                        
                        if let Err(e) = write.send(Message::Text(unsubscribe_msg.to_string())).await {
                            eprintln!("Failed to send unsubscribe message: {}", e);
                            break;
                        }
                    }
                    WebSocketMessage::Close => {
                        if let Err(e) = write.close().await {
                            eprintln!("Failed to close WebSocket: {}", e);
                        }
                        break;
                    }
                }
            }
        });

        // 启动消息接收循环
        tokio::spawn(async move {
            tracing::info!("[WEBSOCKET] 消息接收循环已启动");
            while let Some(msg) = read.next().await {
                match msg {
                    Ok(Message::Text(text)) => {
                        tracing::debug!("[WEBSOCKET] 收到文本消息: {}", text);
                        if let Ok(ws_message) = serde_json::from_str::<WebSocketStreamMessage>(&text) {
                            // 处理WebSocket消息
                            match ws_message {
                                WebSocketStreamMessage::Ticker(ticker) => {
                                    tracing::info!("[WEBSOCKET] 价格更新: {} = {}", ticker.symbol, ticker.price);
                                }
                                WebSocketStreamMessage::Depth(depth) => {
                                    tracing::info!("[WEBSOCKET] 订单簿更新: {}", depth.symbol);
                                }
                                WebSocketStreamMessage::Kline(kline) => {
                                    tracing::info!("[WEBSOCKET] K线更新: {} {}", kline.symbol, kline.close);
                                }
                                WebSocketStreamMessage::Trade(trade) => {
                                    tracing::info!("[WEBSOCKET] 交易更新: {} {} {}", trade.symbol, trade.price, trade.quantity);
                                }
                                WebSocketStreamMessage::OrderUpdate(order) => {
                                    tracing::info!("[WEBSOCKET] 订单更新: {} {}", order.symbol, order.order_status);
                                }
                                WebSocketStreamMessage::AccountUpdate(account) => {
                                    tracing::info!("[WEBSOCKET] 账户更新: {} 余额", account.balances.len());
                                }
                                WebSocketStreamMessage::Error(error) => {
                                    tracing::error!("[WEBSOCKET] 错误: {}", error.msg);
                                }
                            }
                        } else {
                            tracing::debug!("[WEBSOCKET] 无法解析消息: {}", text);
                        }
                    }
                    Ok(Message::Close(_)) => {
                        tracing::info!("[WEBSOCKET] 连接已关闭");
                        break;
                    }
                    Err(e) => {
                        tracing::error!("[WEBSOCKET] 错误: {}", e);
                        break;
                    }
                    _ => {
                        tracing::debug!("[WEBSOCKET] 收到其他类型消息");
                    }
                }
            }
            tracing::info!("[WEBSOCKET] 消息接收循环已结束");
        });

        Ok(())
    }

    /// 订阅实时价格
    pub async fn subscribe_ticker(&mut self, symbol: &str) -> Result<()> {
        let stream = format!("{}@ticker", symbol.to_lowercase());
        self.subscribe_stream(&stream).await
    }

    /// 订阅订单簿
    pub async fn subscribe_depth(&mut self, symbol: &str, levels: u32) -> Result<()> {
        let stream = format!("{}@depth{}", symbol.to_lowercase(), levels);
        self.subscribe_stream(&stream).await
    }

    /// 订阅K线数据
    pub async fn subscribe_kline(&mut self, symbol: &str, interval: &str) -> Result<()> {
        let stream = format!("{}@kline_{}", symbol.to_lowercase(), interval);
        self.subscribe_stream(&stream).await
    }

    /// 订阅交易流
    pub async fn subscribe_trades(&mut self, symbol: &str) -> Result<()> {
        let stream = format!("{}@trade", symbol.to_lowercase());
        self.subscribe_stream(&stream).await
    }

    /// 订阅用户数据流
    pub async fn subscribe_user_data(&mut self, listen_key: &str) -> Result<()> {
        let stream = format!("{}", listen_key);
        self.subscribe_stream(&stream).await
    }

    /// 订阅多个流
    pub async fn subscribe_multiple_streams(&mut self, streams: &[String]) -> Result<()> {
        for stream in streams {
            self.subscribe_stream(stream).await?;
        }
        Ok(())
    }

    /// 取消订阅
    pub async fn unsubscribe(&mut self, stream: &str) -> Result<()> {
        if let Some(sender) = &self.sender {
            sender.send(WebSocketMessage::Unsubscribe(stream.to_string()))
                .map_err(|e| BinanceError::NetworkError(format!("Failed to send unsubscribe message: {}", e)))?;
        }
        Ok(())
    }

    /// 关闭连接
    pub async fn close(&mut self) -> Result<()> {
        if let Some(sender) = &self.sender {
            sender.send(WebSocketMessage::Close)
                .map_err(|e| BinanceError::NetworkError(format!("Failed to send close message: {}", e)))?;
        }
        Ok(())
    }

    /// 订阅流
    async fn subscribe_stream(&mut self, stream: &str) -> Result<()> {
        tracing::info!("[WEBSOCKET] 订阅流: {}", stream);
        if let Some(sender) = &self.sender {
            sender.send(WebSocketMessage::Subscribe(stream.to_string()))
                .map_err(|e| BinanceError::NetworkError(format!("Failed to send subscribe message: {}", e)))?;
            
            self.streams.push(stream.to_string());
            tracing::info!("[WEBSOCKET] 订阅请求已发送: {}", stream);
        } else {
            tracing::error!("[WEBSOCKET] 发送器不可用");
        }
        Ok(())
    }

}

/// 直接连接WebSocket
async fn connect_direct(
    url: &Url,
) -> std::result::Result<WebSocketStream<tokio_native_tls::TlsStream<TcpStream>>, BinanceError> {
    let domain = url.domain()
        .ok_or_else(|| BinanceError::ConfigError("Invalid domain".to_string()))?;
    let port = url.port_or_known_default().unwrap_or(443);
    
    tracing::info!("[WEBSOCKET] 目标域名: {}, 端口: {}", domain, port);
    
    // 建立TCP连接
    let tcp_stream = TcpStream::connect((domain, port)).await
        .map_err(|e| BinanceError::NetworkError(format!("Failed to connect to {}:{}: {}", domain, port, e)))?;
    
    // TLS升级
    tracing::info!("[WEBSOCKET] 建立TLS连接...");
    let native_tls = native_tls::TlsConnector::new()
        .map_err(|e| BinanceError::NetworkError(format!("Failed to create TLS connector: {}", e)))?;
    let tls_connector = TlsConnector::from(native_tls);
    let tls_stream = tls_connector.connect(domain, tcp_stream).await
        .map_err(|e| BinanceError::NetworkError(format!("Failed to establish TLS connection: {}", e)))?;
    
    // 建立WebSocket连接
    tracing::info!("[WEBSOCKET] 建立WebSocket连接...");
    let (ws_stream, _) = client_async(url, tls_stream).await
        .map_err(|e| BinanceError::NetworkError(format!("WebSocket connection failed: {}", e)))?;
    
    Ok(ws_stream)
}

/// 通过代理连接WebSocket
async fn connect_through_proxy(
    url: &Url,
    proxy: &str,
) -> std::result::Result<WebSocketStream<tokio_native_tls::TlsStream<TcpStream>>, BinanceError> {
    let domain = url.domain()
        .ok_or_else(|| BinanceError::ConfigError("Invalid domain".to_string()))?;
    let port = url.port_or_known_default().unwrap_or(443);
    
    tracing::info!("[WEBSOCKET] 目标域名: {}, 端口: {}", domain, port);
    
    // 解析代理地址
    let proxy_addr = if proxy.starts_with("http://") {
        &proxy["http://".len()..]
    } else {
        proxy
    };
    
    tracing::info!("[WEBSOCKET] 代理地址: {}", proxy_addr);
    
    // 建立到代理的TCP连接
    let mut stream = TcpStream::connect(proxy_addr).await
        .map_err(|e| BinanceError::NetworkError(format!("Failed to connect to proxy: {}", e)))?;
    
    // 发送CONNECT请求到代理
    let connect_req = format!(
        "CONNECT {}:{} HTTP/1.1\r\nHost: {}:{}\r\n\r\n",
        domain, port, domain, port
    );
    
    tracing::info!("[WEBSOCKET] 发送CONNECT请求: {}", connect_req.trim());
    stream.write_all(connect_req.as_bytes()).await
        .map_err(|e| BinanceError::NetworkError(format!("Failed to send CONNECT request: {}", e)))?;
    
    // 读取代理响应
    let mut buf = [0u8; 1024];
    let n = stream.read(&mut buf).await
        .map_err(|e| BinanceError::NetworkError(format!("Failed to read proxy response: {}", e)))?;
    let resp = String::from_utf8_lossy(&buf[..n]);
    
    tracing::info!("[WEBSOCKET] 代理响应: {}", resp.trim());
    
    if !resp.contains("200") {
        return Err(BinanceError::NetworkError(format!("Proxy CONNECT failed: {}", resp)));
    }
    
    // TLS升级
    tracing::info!("[WEBSOCKET] 建立TLS连接...");
    let native_tls = native_tls::TlsConnector::new()
        .map_err(|e| BinanceError::NetworkError(format!("Failed to create TLS connector: {}", e)))?;
    let tls_connector = TlsConnector::from(native_tls);
    let tls_stream = tls_connector.connect(domain, stream).await
        .map_err(|e| BinanceError::NetworkError(format!("Failed to establish TLS connection: {}", e)))?;
    
    // 建立WebSocket连接
    tracing::info!("[WEBSOCKET] 建立WebSocket连接...");
    let (ws_stream, _) = client_async(url, tls_stream).await
        .map_err(|e| BinanceError::NetworkError(format!("WebSocket connection failed: {}", e)))?;
    
    Ok(ws_stream)
}
